package framework;
import com.fasterxml.jackson.core.JsonProcessingException;
import util.StringUtil;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.ResourceAlreadyExistsException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryAction;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.sort.SortBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;

/**
 * Created by Administrator on 2017/9/28.
 */
//https://www.elastic.co/guide/en/elasticsearch/client/java-api/current/index.html
public class ElasticSearchOpTemplate implements ElasticSearchTemplate {

    public Logger logger = LoggerFactory.getLogger(ElasticSearchOpTemplate.class);

    private IndexClient indexClient;

    private MappingClient mappingClient;

    public ElasticSearchOpTemplate(){
        indexClient = new IndexClient();
    }



    //添加一个文档
    //{"id":"2","name":"3":}

    private  boolean createIndex(String indexName){
        logger.info("ElasticSearchOpTemplate create  indexName:"+indexName);
        return  indexClient.createIndex(indexName);
    }

    public  boolean delIndex(String indexName){
        return indexClient.delIndex(indexName);
    }


    private  boolean isIndexExists( String indexName) {
        return indexClient.isIndexExists(indexName);
    }

    private  boolean isTypeExists(String indexName, String typeName)
            throws InterruptedException, ExecutionException {
        return indexClient.isTypeExists(indexName,typeName);
    }

    //只有在导入的时候才调用创建mapping
    @Override
    public boolean createMapping(String indexName, String type, String mappingJson) throws Exception {
        if(isIndexExists(indexName)){
            logger.info("createMapping indexName:"+indexName+",has exists");
            if(isTypeExists(indexName,type)){
                logger.info("createMapping type:"+type+",has exists");
            }
            return false;
        }

        //3添加（多台tomcat）
        try{
            //多个现在走这步,如果索引存在就报异常 ResourceAlreadyExistsException
            createIndex(indexName);
            //mapping会覆盖，不会报错，
            mappingClient.createOrUpdateMapping(indexName,type,mappingJson);
        }catch (ResourceAlreadyExistsException e){
            //如果有多个创建就会报错吗
            //throw  new RuntimeException();
            Thread.sleep(200);//200ms的等待异常
            if(!isTypeExists(indexName,type)){
                logger.info("createMapping ResourceAlreadyExistsException exception recreate mapping:"+type);
                mappingClient.createOrUpdateMapping(indexName,type,mappingJson);
            }
        }
        logger.info("ElasticSearchOpTemplate createOrUpdateMapping:"+type);
        return true;
    }

    private void checkIndexAndMapping(String indexName,String typeMapping)  {
        if(!isIndexExists(indexName)){
            throw new RuntimeException("indexName not exist");
        }
        try {
            if(!isTypeExists(indexName,typeMapping)){
                throw new RuntimeException("typeMapping must create");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     *
     * @param indexName 数据库名称
     * @param typeName 表名
     * @param primaryKey 表的主键字段，最好是Uuid
     * @param jsonData
     */
    @Override
    public  void addOrUpdate(String indexName,String typeName,String primaryKey,String jsonData){
        //checkIndexAndMapping(indexName,typeName);
        if(StringUtil.isNotBlank(primaryKey)){
            indexClient.getClient().prepareIndex(indexName, typeName,primaryKey)
                    .setSource(jsonData, XContentType.JSON)
                            //.setId(jsonData.getString(KEY))//自己设置了id，也可以使用ES自带的，但是看文档说，ES的会因为删除id发生变动。
                    .execute()
                    .actionGet();
        }else{
            indexClient.getClient().prepareIndex(indexName, typeName)
                    .setSource(jsonData, XContentType.JSON)
                            //.setId(jsonData.getString(KEY))//自己设置了id，也可以使用ES自带的，但是看文档说，ES的会因为删除id发生变动。
                    .execute()
                    .actionGet();
        }

    }

    @Override
    public String batchAddOrUpdate(List<Map<String, Object>> list, String indexName, String typeName, String primaryKeyName) throws JsonProcessingException {
        //checkIndexAndMapping(indexName,typeName);
        if (CollectionUtils.isEmpty(list)) {
            return null;
        }
        Client client =  indexClient.getClient();
        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
        for (Map<String, Object> map : list) {
            String id = String.valueOf(map.get(primaryKeyName));
            bulkRequestBuilder.add(client.prepareIndex(indexName, typeName, id).setSource(map));
        }
        BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
        bulkRequestBuilder.request().requests().clear();
        if (bulkResponse.hasFailures()) {
            // return bulkResponse.buildFailureMessage();
            BulkItemResponse[] items = bulkResponse.getItems();
            String indexIds = "";
            for (int i = 0; i < items.length; ++i) {
                BulkItemResponse response = items[i];
                if (response.isFailed()) {
                    indexIds += ("," + response.getId());
                }
            }
            logger.error("es index import failure:" + bulkResponse.buildFailureMessage() + "[" + indexName + "][" + typeName + "][" + indexIds + "]");
            return StringUtils.isBlank(indexIds) ? null : indexIds.substring(1);
        }
        return null;
    }

    /**
     * 异步删除
     * @param indexName
     * @param typeName
     * @param primarykey
     * @return
     */
    private void deleteByPrimakryAsyn(String indexName,String typeName,String primarykey){
        DeleteRequest request = new DeleteRequest(indexName,typeName,primarykey); //
        indexClient.getClient().delete(request, new ActionListener<DeleteResponse>() { //
            @Override
            public void onResponse(DeleteResponse deleteResponse) {
                //这里做逻辑
            }

            @Override
            public void onFailure(Exception e) {
                //
            }
        });
        //这里写其他主线程逻辑
    }

    /**
     * 删除某个表的某行数据
     * @param indexName
     * @param typeName
     * @param primaryKey
     * @return
     */
    @Override
    public boolean deleteByPrimaryKey(String indexName,String typeName,String primaryKey){
        DeleteRequest  request = new DeleteRequest(indexName,typeName,primaryKey); //
        DeleteResponse response = indexClient.getClient().delete(request).actionGet(); //
        return response.getResult() == DocWriteResponse.Result.DELETED;
    }

    @Override
    public boolean deleteByPrimaryKeyList(String indexName, String typeName, List<String> primaryKeyList) {
        BulkRequestBuilder bulkRequest = indexClient.getClient().prepareBulk();
        for(String id : primaryKeyList){
            bulkRequest.add(indexClient.getClient().prepareDelete(indexName, typeName, id).request());
        }
        BulkResponse bulkResponse = bulkRequest.get();
        if (bulkResponse.hasFailures()) {
            for(BulkItemResponse item : bulkResponse.getItems() ){
                if(item.isFailed()){
                    logger.error("deleteByPrimaryKeyList id=" + item.getId() + ",message==" + item.getFailureMessage());
                }
            }
            return false;
        } else {
           return true;
        }
    }

//    private void batchDelete(){
//        String index = "wareic";
//        String type = "product";
//        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setTypes(type);
//        searchRequestBuilder.setFrom(0).setSize(1000);
//        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
//        queryBuilder.must(QueryBuilders.termQuery("groupID", "3445"));
//        searchRequestBuilder.setQuery(queryBuilder);
//        SearchResponse response = searchRequestBuilder.execute().get();
//        for(SearchHit hit : response.getHits()){
//            String id = hit.getId();
//
//        }
//        BulkResponse bulkResponse = bulkRequest.get();
//        if (bulkResponse.hasFailures()) {
//            for(BulkItemResponse item : bulkResponse.getItems()){
//                System.out.println(item.getFailureMessage());
//            }
//        } else {
//            System.out.println("delete ok");
//        }
//    }

    @Override
    public boolean deleteByCondition(String indexName, String typeName, QueryBuilder deleteContidionBuilder) {
            try{
                BulkByScrollResponse response= DeleteByQueryAction.INSTANCE.newRequestBuilder(indexClient.getClient())
                        .abortOnVersionConflict(false)
                        .filter(deleteContidionBuilder).source(indexName).get();
                long delNum =  response.getDeleted();
                logger.info("deleteByCondition_indeName:"+indexName+",del data:"+delNum);
                return  true;
            }catch (Exception e){
                e.printStackTrace();
            }

//        SearchRequestBuilder searchRequestBuilder =indexClient.getClient().prepareSearch(indexName).setTypes(typeName);
//        SearchRequest searchRequest = searchRequestBuilder
//                .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
//                .setQuery(deleteContidionBuilder).request();
//        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(searchRequest);
//        deleteByQueryRequest.indices(indexName);
//        deleteByQueryRequest.types(typeName);
        return true;

    }

    /**
     * TODO 根据主键查询单条记录********************************
     *
     * @param indexName 索引名称
     * @param type      索引类型
     * @return
     */
    @Override
    public Map<String, Object> getSingleRecordByPrimaryKey(String indexName, String type, String primaryKey) {
        Client client =indexClient.getClient();
        GetResponse searchResponse = client.prepareGet(indexName, type, primaryKey)
                .setOperationThreaded(false)
                .execute()
                .actionGet();
        return searchResponse.getSource();
    }

    /**
     * 查询记录
     * @param queryBuilder createSearchRequestBuilder 构建一个查询
     * @indexName 数据库名
     * @type      表名
     * @param startCount 其实偏移量
     * @param record size数据
     * @return
     */
    @Override
    public  EsWrapperSearchResponse searchRecordList(QueryBuilder  queryBuilder, String indexName,String type,int startCount, int record,SortBuilder... sortBuilderList) {
        List<Map<String, Object>> resultList = new ArrayList<Map<String, Object>>();
        long start = new Date().getTime();
        Client client =indexClient.getClient();
        SearchRequestBuilder searchRequestBuilder =client.prepareSearch(indexName).setTypes(type);

        if(sortBuilderList!=null){
            for(SortBuilder sTemp : sortBuilderList){
                if(sTemp!=null){
                    searchRequestBuilder.addSort(sTemp);
                }
            }
        }
        //and搜索条件
        SearchResponse searchResponse =searchRequestBuilder
                .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                .setFrom(startCount).setSize(record).setExplain(false)
                .setQuery(queryBuilder)
                .execute()
                .actionGet();
        SearchHits hits = searchResponse.getHits();
        long end = new Date().getTime();
        long queryCount = hits.getHits().length;
        long allCount = hits.getTotalHits();
        if (queryCount < record) {
            record = (int) queryCount;
        }
        for (int i = 0; i < record; i++) {
            resultList.add(hits.getAt(i).getSource());
        }
        EsWrapperSearchResponse returnResult = new EsWrapperSearchResponse();
        returnResult.setList(resultList);
        returnResult.setCount(allCount);
        return returnResult;

    }

    @Override
    public SearchResponse searchRecordListToNativeResult(QueryBuilder queryBuilder, String indexName, String type, int startCount, int record,AggregationBuilder aggregationBuilder, SortBuilder... sortBuilderList) {
        List<Map<String, Object>> resultList = new ArrayList<Map<String, Object>>();
        long start = new Date().getTime();
        Client client =indexClient.getClient();
        SearchRequestBuilder searchRequestBuilder =client.prepareSearch(indexName).setTypes(type);

        if(sortBuilderList!=null){
            for(SortBuilder sTemp : sortBuilderList){
                if(sTemp!=null){
                    searchRequestBuilder.addSort(sTemp);
                }
            }
        }

        if(aggregationBuilder!=null){
            searchRequestBuilder.addAggregation(aggregationBuilder);
        }
        //and搜索条件
        SearchResponse searchResponse =searchRequestBuilder
                .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                .setFrom(startCount).setSize(record).setExplain(false)
                .setQuery(queryBuilder)
                .execute()
                .actionGet();
        return  searchResponse;
    }

    //搜索例子
    public  void searchOrder(Client client,String indexName,String type){
//        SearchRequestBuilder srb1 = client
//                .prepareSearch().setQuery(QueryBuilders.queryStringQuery("elasticsearch")).setFrom(0).setSize(100);
        QueryBuilder builder12 = QueryBuildersWrapper.termQuery("tradeStatus", "有效");
        QueryBuilder builder3 =  QueryBuildersWrapper.termQuery("propertyId",390168);
        QueryBuilder builder31 = QueryBuildersWrapper.termQuery("estateName","上南三村");

        BoolQueryBuilder boolQueryBuilder =QueryBuildersWrapper.boolQuery();
        boolQueryBuilder.must(builder12);

    }

}
